package cn.itcast.streaming.function.flatmap;

import cn.itcast.streaming.bean.ElectricFenceModel;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

//对电子围栏分析结果继续拉宽(uuid和inMysql)
public class ElectricFenceModelFunction implements CoFlatMapFunction<ElectricFenceModel, HashMap<String, Long>, ElectricFenceModel> {
    private HashMap<String,Long> electricFenceModelMapResult = new HashMap<>();
    //对分析结果模型数据操作
    @Override
    public void flatMap1(ElectricFenceModel electricFenceModel, Collector<ElectricFenceModel> collector) throws Exception {
        System.out.println("尝试最终的窗口拉宽操作：");
        if(electricFenceModelMapResult.getOrDefault(electricFenceModel.getVin(),null) != null){
            //mysql中存在进入电子围栏的数据，但是出电子围栏时间为空的数据，因此需要将出电子围栏时间更新掉
            Long electricFenceId = electricFenceModelMapResult.get(electricFenceModel.getVin());
            electricFenceModel.setUuid(electricFenceId);
            electricFenceModel.setInMysql(true);
            collector.collect(electricFenceModel);
        }else {
            electricFenceModel.setUuid(Long.MAX_VALUE);
            electricFenceModel.setInMysql(false);
            collector.collect(electricFenceModel);
        }
    }

    //对分析结果表数据操作
    @Override
    public void flatMap2(HashMap<String, Long> stringLongHashMap, Collector<ElectricFenceModel> collector) throws Exception {
        electricFenceModelMapResult = stringLongHashMap;
        System.out.println("hash:"+electricFenceModelMapResult);
    }
}
